package sf.config.reactive;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.lang.NonNull;
import org.springframework.transaction.support.TransactionCallback;
import org.springframework.transaction.support.TransactionTemplate;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Scheduler;

import java.util.concurrent.Callable;

public class ReactiveTransactionTemplate {
    private Logger LOG = LoggerFactory.getLogger(ReactiveTransactionTemplate.class);


    @NonNull
    private final TransactionTemplate transactionTemplate;

    private Scheduler scheduler;

    public ReactiveTransactionTemplate(@NonNull TransactionTemplate transactionTemplate, Scheduler scheduler) {
        this.transactionTemplate = transactionTemplate;
        this.scheduler = scheduler;
    }

    @NonNull
    public <T> Mono<T> execute(@NonNull TransactionCallback<T> action) {
//        Scheduler scheduler = Schedulers.newSingle(new DefaultThreadFactory("transaction"));
//        return Mono.fromCallable(() -> transactionTemplate.execute(action)).publishOn(scheduler).doFinally(t->scheduler.dispose());
        return Mono.fromCallable(() -> transactionTemplate.execute(action)).subscribeOn(scheduler);
    }

    @NonNull
    public <T> Mono<T> async(Callable<T> callable) {
//        Scheduler scheduler = Schedulers.newSingle(new DefaultThreadFactory("transaction"));
//        return Mono.fromCallable(callable).publishOn(scheduler).doFinally(t -> scheduler.dispose());
        return Mono.fromCallable(callable).subscribeOn(scheduler);
    }
}